package monitor

import (
	"context"
	"encoding/json"
	"time"

	"code.byted.org/bytertc/whiteboard_server/biz/service/transcode/task"
	"code.byted.org/videoarch/universal_utils_go/event_collector/tracer"
	"code.byted.org/videoarch/universal_utils_go/event_collector/util"
	"github.com/spf13/cast"

	"code.byted.org/bytertc/whiteboard_server/biz/dal"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/gateway"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/mmetrics"
	uutil "code.byted.org/bytertc/whiteboard_server/biz/pkg/util"
	"code.byted.org/gopkg/logs/v2/log"
	"code.byted.org/gopkg/metrics"
)

func NewMonitor(repo dal.Repo, rtcAPI gateway.RtcAPI, metricsClient *metrics.MetricsClientV2) *Monitor {
	return &Monitor{
		repo:          repo,
		rtcAPI:        rtcAPI,
		metricsClient: metricsClient,
	}
}

type Monitor struct {
	repo          dal.Repo
	rtcAPI        gateway.RtcAPI
	metricsClient *metrics.MetricsClientV2
}

func (m *Monitor) Start() {

	go func(m *Monitor) {
		defer uutil.CheckPanic()
		ctx := context.Background()
		ticker := time.NewTicker(time.Second)
		for range ticker.C {
			if err := m.repo.LockStaticMonitor(ctx); err != nil {
				continue
			}
			log.V1.CtxInfo(ctx, "Monitor static")
			tasks, err := m.repo.GetExceptionStaticTask(ctx)
			if err != nil {
				log.V1.CtxError(ctx, "Monitor GetExceptionStaticTask Error:%v", err)
				continue
			}
			if len(tasks) == 0 {
				continue
			}
			taskIDs := make([]string, 0)
			for _, t := range tasks {
				taskIDs = append(taskIDs, t.TaskId)
			}
			err = m.repo.UpdateExceptionStaticTask(ctx, taskIDs)
			if err != nil {
				log.V1.CtxError(ctx, "Monitor UpdateExceptionStaticTask error:%v", err)
			}
			for _, t := range tasks {
				if err = m.repo.LockTranscodeTask(ctx, t.TaskId); err != nil {
					continue
				}
				tt := &task.Task{}
				err = json.Unmarshal([]byte(t.Meta), tt)
				if err != nil {
					continue
				}
				traceKey := &util.KeyWord{
					AppId:  t.AppId,
					UserId: t.UserId,
					TaskId: t.TaskId,
				}
				otherInfo := map[string]string{
					"transcode_mode": cast.ToString(task.Statistic),
					"input_format":   tt.TranscodeConfig.GetInputFormat(),
					"output_format":  tt.TranscodeConfig.GetOutputFormat(),
					"priority":       cast.ToString(tt.Priority),
					"size":           cast.ToString(t.FileSize),
					"total_pages":    cast.ToString(t.TotalPages),
					"cur_pages":      cast.ToString(t.TotalPages),
					"result":         "fail",
					"fail_reason":    "500",
				}
				tracer.MileStone(tracer.STREAM_COLLECTOR, "transcode_finish", "", traceKey, otherInfo)
				tags := []metrics.T{
					{Name: "app_id", Value: t.AppId},
					{Name: "task_id", Value: t.TaskId},
					{Name: "executor", Value: t.Executor},
				}
				err = m.metricsClient.EmitCounter(mmetrics.MetricsStaticTranscodeException, 1, tags...)
				if err != nil {
					log.V1.CtxError(ctx, "Monitor EmitCounter failed,error:%v", err)
				}
			}
		}
	}(m)

	go func(m *Monitor) {
		defer uutil.CheckPanic()
		ctx := context.Background()
		ticker := time.NewTicker(6 * time.Second)
		for range ticker.C {
			if err := m.repo.LockDynamicMonitor(ctx); err != nil {
				continue
			}
			tasks, err := m.repo.GetExceptionDynamicTask(ctx)
			if err != nil {
				log.V1.CtxError(ctx, "Monitor GetExceptionDynamicTask error:%v", err)
				continue
			}
			taskIDs := make([]string, 0)
			for _, t := range tasks {
				taskIDs = append(taskIDs, t.TaskId)
			}
			if len(taskIDs) == 0 {
				continue
			}
			err = m.repo.UpdateExceptionDynamicTask(ctx, taskIDs)
			if err != nil {
				log.V1.CtxError(ctx, "Monitor UpdateExceptionDynamicTask error:%v", err)
			}
			for _, t := range tasks {
				if err = m.repo.LockTranscodeTask(ctx, t.TaskId); err != nil {
					continue
				}
				tt := &task.Task{}
				err = json.Unmarshal([]byte(t.Meta), tt)
				if err != nil {
					continue
				}
				traceKey := &util.KeyWord{
					AppId:  t.AppId,
					UserId: t.UserId,
					TaskId: t.TaskId,
				}
				otherInfo := map[string]string{
					"transcode_mode": cast.ToString(task.Dynamic),
					"input_format":   "",
					"output_format":  "",
					"priority":       cast.ToString(tt.Priority),
					"size":           cast.ToString(t.FileSize),
					"total_pages":    cast.ToString(t.TotalPages),
					"cur_pages":      cast.ToString(t.TotalPages),
					"result":         "fail",
					"fail_reason":    "500",
				}
				tracer.MileStone(tracer.STREAM_COLLECTOR, "transcode_finish", "", traceKey, otherInfo)
				tags := []metrics.T{
					{Name: "app_id", Value: t.AppId},
					{Name: "task_id", Value: t.TaskId},
					{Name: "file_id", Value: t.FileID},
				}
				err = m.metricsClient.EmitCounter(mmetrics.MetricsDynamicTranscodeException, 1, tags...)
				if err != nil {
					log.V1.CtxError(ctx, "Monitor EmitCounter failed,error:%v", err)
				}
			}
		}
	}(m)
}
